#ifndef __MQ_CHANNEL_
#define __MQ_CHANNEL_
// 所以都指定了头文件查找路径了，就没必要多此一举再用相对路径了，同时也是为了和库统一，所以这里修改为muduo开头路径
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_tools.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "mq_consumer.hpp"
#include "mq_virtualhost.hpp"
#include "mq_route.hpp"
#include <mutex>

namespace MQ
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &cid,
                const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec,
                const ConsumerManager::ptr &cmanager_ptr,
                const VirtualHost::ptr &vp,
                const ThreadPool::ptr &tp)
            : _cid(cid), _conn(conn), _codec(codec), _cmanager_ptr(cmanager_ptr), _vp(vp), _tp(tp)
        {
        }

        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                _cmanager_ptr->remove(_consumer->_tag, _consumer->_qname);
            }
        }

        // 交换机的声明与删除
        void declareExchange(const declareExchangeRequestPtr &derp)
        {
            bool ret = _vp->declareExchange(derp->exchange_name(), derp->exchange_type(), derp->durable(), derp->auto_delete(), derp->args());
            basicResponse(derp->cid(), derp->rid(), ret);
        }

        void deleteExchange(const deleteExchangeRequestPtr &derp)
        {
            _vp->deleteExchange(derp->exchange_name());
            basicResponse(derp->cid(), derp->rid(), true);
        }
        // 队列的声明与删除
        void declareQueue(const declareQueueRequestPtr &dqrp)
        {
            bool ret = _vp->declareQueue(dqrp->queue_name(), dqrp->durable(), dqrp->exclusive(), dqrp->auto_delete(), dqrp->args());
            if (ret == false)
            {
                basicResponse(dqrp->cid(), dqrp->rid(), ret);
                return;
            }
            _cmanager_ptr->initQueueConsumer(dqrp->queue_name());
            basicResponse(dqrp->cid(), dqrp->rid(), ret);
        }

        void deleteQueue(const deleteQueueRequestPtr &dqrp)
        {
            _cmanager_ptr->destoryQueueConsumer(dqrp->queue_name());
            _vp->deleteQueue(dqrp->queue_name());
            basicResponse(dqrp->cid(), dqrp->rid(), true);
        }
        // 队列的绑定和解绑
        void queueBind(const queueBindRequestPtr &qbrp)
        {
            bool ret = _vp->bind(qbrp->exchange_name(), qbrp->queue_name(), qbrp->binding_key());
            basicResponse(qbrp->cid(), qbrp->rid(), ret);
        }

        void queueUnBind(const queueUnBindRequestPtr &qurp)
        {
            _vp->unbind(qurp->exchange_name(), qurp->queue_name());
            basicResponse(qurp->cid(), qurp->rid(), true);
        }

        // 消息的发布和确定
        void basicPublish(const basicPublishRequestPtr &bprp)
        {
            // 1.判断交换机是否存在
            auto ep = _vp->selectExchange(bprp->exchange_name());
            if (ep.get() == nullptr)
            {
                basicResponse(bprp->cid(), bprp->rid(), false);
                return;
            }
            // 2.判断路由
            MsgQueueBindingMap mqbm = _vp->getExchangeBinding(bprp->exchange_name());
            BasicProperties *bp = nullptr;
            std::string routing_key;
            if (bprp->has_properties())
            {
                bp = bprp->mutable_properties();
                routing_key = bp->routing_key();
            }
            for (auto &binding : mqbm)
            {
                if (Router::route(ep->type, routing_key, binding.second->_binding_key))
                {
                    // 3.消息添加队列
                    _vp->basicPublish(binding.first, bp, bprp->body());
                    // 4.添加消息消费任务
                    auto func = std::bind(&Channel::consume, this, binding.first);
                    _tp->push(func);
                }
            }
            basicResponse(bprp->cid(), bprp->rid(), true);
        }

        void basicAck(const basicAckRequestPtr &barp)
        {
            _vp->basicAck(barp->queue_name(), barp->message_id());
            basicResponse(barp->cid(), barp->rid(), true);
        }
        // 订阅消息和取消订阅
        void basicConsume(const basicConsumeRequestPtr &bcrp)
        {
            // 1. 判断队列是否存在
            bool ret = _vp->existsQueue(bcrp->queue_name());
            if (ret == false)
            {
                basicResponse(bcrp->cid(), bcrp->rid(), false);
                return;
            }
            // 2. 创建消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            _consumer = _cmanager_ptr->create(bcrp->consumer_tag(), bcrp->queue_name(), bcrp->auto_ack(), cb);
            basicResponse(bcrp->cid(), bcrp->rid(), true);
        }

        void basicCancel(const basicCancelRequestPtr &berp)
        {
            _cmanager_ptr->remove(berp->consumer_tag(), berp->queue_name());
            basicResponse(berp->cid(), berp->rid(), true);
        }

    private:
        void basicResponse(const std::string &cid, const std::string &rid, bool ok)
        {
            basicCommonResponse resp;
            resp.set_cid(cid);
            resp.set_rid(rid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }

        void callback(const std::string tag, const BasicProperties *bp, const std::string &body)
        {
            basicConsumeResponse bcrp;
            bcrp.set_cid(_cid);
            bcrp.set_body(body);
            bcrp.set_consumer_tag(tag);
            if (bp)
            {
                bcrp.mutable_properties()->set_id(bp->id());
                bcrp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                bcrp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, bcrp);
        }

        void consume(const std::string &qname)
        {
            // 从队列取出消息
            MessagePtr mp = _vp->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                LOG(ERROR, "执行消费任务失败，%s 队列没有消息！", qname.c_str());
                return;
            }
            // 从队列取出消费者
            Consumer::ptr cp = _cmanager_ptr->RR(qname);
            if (cp.get() == nullptr)
            {
                LOG(ERROR, "执行消费任务失败，%s 队列没有消费者！", qname.c_str());
                return;
            }
            // 调用消费处理函数
            cp->_cb(cp->_tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            // 判断是否自动应答
            if (cp->_auto_ack)
                _vp->basicAck(qname, mp->payload().properties().id());
        }

    private:
        std::string _cid;
        Consumer::ptr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ConsumerManager::ptr _cmanager_ptr;
        VirtualHost::ptr _vp;
        ThreadPool::ptr _tp;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager()
        {
        }

        bool openChannel(const std::string &id,
                         const VirtualHost::ptr &vp,
                         const ConsumerManager::ptr &cmp,
                         const ProtobufCodecPtr &codec,
                         const muduo::net::TcpConnectionPtr &conn,
                         const ThreadPool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it != _channels.end())
            {
                LOG(WARNING, "信道：%s 已经存在!", id.c_str());
                return false;
            }
            auto channel = std::make_shared<Channel>(id, conn, codec, cmp, vp, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }

        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }

        Channel::ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}

#endif